package org.example.utils;

import lombok.extern.slf4j.Slf4j;
import org.example.bean.enums.SSEMsgType;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

@Slf4j
public class SSEServer {

    public static Map<String, SseEmitter> sseClients = new ConcurrentHashMap<>();

    private static AtomicInteger onlineCount= new AtomicInteger(0);

    public static SseEmitter connect(String userId){
        SseEmitter sseEmitter = new SseEmitter(0L);

        sseEmitter.onTimeout(timeoutCallBack(userId));
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        log.info("创建SSE连接，用户id是{}",userId);

        sseClients.put(userId, sseEmitter);

       // onlineCount.getAndIncrement();

        return sseEmitter;
    }

    private static Runnable completionCallBack(String userId){
        return () -> {
            log.info("用户{}的SSE连接已完成",userId);
            removeSseClient(userId);
        };
    }

    private static Consumer<Throwable> errorCallBack(String userId){
        return throwable -> {
            log.error("用户{}的SSE连接已出错: {}", userId, throwable.getMessage());
            removeSseClient(userId);
        };
    }

    private static Runnable timeoutCallBack(String userId){
        return () -> {
            log.info("用户{}的SSE连接已超时",userId);
            removeSseClient(userId);
        };
    }

    public static void stop(String userId){
        if (!sseClients.containsKey(userId)){
            log.info("用户{}的SSE连接无需关闭",userId);
            return;
        }

        SseEmitter sseEmitter = sseClients.get(userId);
        try {
            sseEmitter.complete();
            sseClients.remove(userId);
            log.info("用户{}的SSE连接已关闭",userId);
        }
        catch (Exception e){
            log.error("用户{}的SSE连接已出错: {}", userId, e.getMessage());
        }
    }

    public static void sendMassage(String userId,String massage, SSEMsgType sseMsgType){
        if (!sseClients.containsKey(userId)){
            return;
        }

        sendMassageSseEmitter(sseClients.get(userId),userId,massage,sseMsgType);
    }

    public static void sendMassageToAllUsers(String massage, SSEMsgType sseMsgType){
        if (sseClients.isEmpty()){
            return;
        }

        sseClients.forEach((userId,sseEmitter) -> sendMassageSseEmitter(sseEmitter,userId,massage,sseMsgType));
    }

    private static void sendMassageSseEmitter(SseEmitter sseEmitter,String userId,String massage, SSEMsgType sseMsgType){
        try {
            SseEmitter.SseEventBuilder msg = SseEmitter.event()
                    .id(userId)
                    .name(sseMsgType.type)
                    .data(massage);
            sseEmitter.send(msg);
        }
        catch (Exception e){
            log.error("用户{}的SSE连接已出错: {}", userId, e.getMessage());
            removeSseClient(userId);
        }
    }

    private static void removeSseClient(String userId){
        SseEmitter sseEmitter = sseClients.get(userId);
        if(sseEmitter != null){
            sseEmitter.complete();
            sseClients.remove(userId);
            onlineCount.getAndDecrement();
            log.info("用户{}的SSE连接已关闭",userId);
        }
    }

    public static int getOnlineCount(){
        return onlineCount.intValue();
    }

}
